package org.platform.lwc.kafka.service;

import cn.hutool.json.JSONUtil;
import org.platform.lwc.kafka.model.KafkaMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

public class KafkaMessageSender {


    private KafkaTemplate<String, String> kafkaTemplate;

    private static final Logger log= LoggerFactory.getLogger(KafkaMessageSender.class);


    public void sendMessage(String topic, KafkaMessage message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, JSONUtil.toJsonStr(message));
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("kafka sendMessage error, ex = {}, topic = {}, data = {}", ex, topic, JSONUtil.toJsonStr(message));
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("kafka sendMessage success topic = {}, data = {}",topic, JSONUtil.toJsonStr(message));
            }
        });
    }

    public KafkaMessageSender(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
}
